package component.hadoop;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.log4j.Logger;

//only conver the Tracking Log to records
//Map output: sessionid + RecordType ==> Info
public class SimpleTracking {
	final static Text emptyTag = new Text("-");
	
	//final static Text separator = new Text("{&&}");
	//final static String delimeter = "|#";

	//change for String.split
	final static Text separator = new Text("\\{\\&\\&\\}");
	final static String delimeter = "\\|#";

	//type + version => List
	static Map<VersionedType, LogFieldList>   logFormatMap = new HashMap<VersionedType, LogFieldList>();
	static Map<String, FieldLogFieldMapping>  logFileTableMap = new LinkedHashMap<String, FieldLogFieldMapping>();

	static Logger logger = Logger.getLogger(SimpleTracking.class);
	static {
		initLogFormatMap();
	}

	static void initLogFormatMap() {
		String[][]  config = new String[][]{
				{"affiliate_outgoing_log","20081027","sessionid|#visittime|#siteid|#productid|#merchantid|#categoryid|#clickarea|#destsite|#channelid|#revenue|#desturl|#randstr|#currandstr"},
				{"affiliate_transfer_log","20081027","sessionid|#visittime|#siteid|#valid|#keyword|#categoryid|#resulttype|#resultcount|#source|#isrealsearch|#iscached|#channelid|#costtime|#totalcosttime|#productid|#responsetime|#resultsize|#randstr|#currandstr"},
				{"client_page_visit_log","20101126","sessionid|#visittime|#siteid|#clienttime|#loadtime|#cookieenabled|#javaenabled|#jsenabled|#screenresolution|#timezone|#languagesetting|#flashversion|#currandstr"},
				{"error_log","20081027","sessionid|#visittime|#siteid|#requesturi|#referer|#remark|#randstr|#currandstr"},
				{"incoming_log","20081027","sessionid|#visittime|#siteid|#visitip|#pagetype|#valid|#requesturi|#httpreferer|#httpuseragent|#httpstatus|#channelid|#categoryid|#productid|#testkey|#userid|#source|#sourcegroup"},
				{"offer_impression_log","20081027","sessionid|#visittime|#siteid|#productid|#merchantid|#bidposition|#totalmerchantcount|#showarea|#displayposition|#channelid|#offerid|#datasource|#sdcofferid|#businesstype|#randstr|#currandstr"},
				{"offer_outgoing_log","20100317","sessionid|#visittime|#siteid|#productid|#merchantid|#businesstype|#offerid|#outgoingtype|#bidposition|#totalmerchantcount|#clickarea|#displayposition|#pricerank|#raterank|#sortby|#channelid|#revenue|#cpcforlogo|#extracpc|#desturl|#datasource|#sdcofferid|#randstr|#currandstr|#outgoingid"},
				{"page_visit_log","20100206","sessionid|#visittime|#siteid|#visitip|#pagetype|#pagevisitorder|#requesturi|#httpreferer|#httpuseragent|#httpstatus|#connectstatus|#channelid|#categoryid|#productid|#randstr|#currandstr|#traffictype"},
				{"search_log","20081027","sessionid|#visittime|#siteid|#valid|#channelid|#categoryid|#productid|#source|#keyword|#matchkeyword|#isrealsearch|#iscached|#searchenginetype|#resulttype|#resultcount|#resultsize|#costtime|#totalcosttime|#responsetime|#desturl|#randstr|#currandstr"},
				{"sponsor_impression_log","20081027","sessionid|#visittime|#siteid|#sponsortype|#channelid|#keyword|#impressionCount|#channeltag|#randstr|#currandstr"},
				{"sponsor_outgoing_log","20100206","sessionid|#visittime|#siteid|#sponsortype|#channelid|#revenue|#keyword|#displayPosition|#advertiserHost|#channeltag|#desturl|#randstr|#currandstr|#outgoingtype"},
				{"sponsor_transfer_log","20100125","sessionid|#visittime|#siteid|#valid|#sponsortype|#channelid|#keyword|#requestip|#resultcount|#costtime|#channeltag|#randstr|#currandstr|#requestcount"},

		};

		for(String[] item: config) {
			LogFieldList list = new LogFieldList();
			//wrong: use | and # as deilimiter
			//String[] fields = StringUtils.splitPreserveAllTokens(item[2],"|#");
			String[] fields = item[2].split("\\|#");
			for(int i =0 ; i < fields.length; i ++) {
				list.put(fields[i], i);
			}
			logFormatMap.put(new VersionedType(item[0], item[1]), list);
		}
	}

	static {
		initLogFileTableMapping();
	}

	static void initLogFileTableMapping() {
		//logname ==> dwname + fields list
		String[][] logDwTable={
				{"incoming_log", "Incoming"},
				{"page_visit_log", "PageVisit"},
				{"sponsor_outgoing_log","Outgoing"},
				{"offer_outgoing_log","Outgoing"},
				{"affiliate_outgoing_log","Outgoing"},
				{"search_log","Search"},
				{"affiliate_transfer_log","Search"},
				{"sponsor_transfer_log","SponsorTransfer"},
				{"sponsor_impression_log","Impression"},
				{"product_impression_log","Impression"},
				{"affiliate_impression_log","Impression"},
				{"offer_impression_log","Impression"},
				{"client_page_visit_log","ClientPV"},
		};

		String[][][] fieldMapping = new String[][][]{
				new String[][] { // [incoming-incoming_log]
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP", "visitip",},
						{"WebSite", "siteid",},
						{"WebPage", "",},
						{"Valid", "valid",},
						{"RequestUri", "requesturi",},
						{"HttpUserAgent", "httpuseragent",},
						{"HttpReferer", "httpreferer",},
						{"HttpStatus", "httpstatus",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"ProductID", "productid",},
						{"TestKey", "testkey",},
						{"UserID", "userid",},
						{"Source", "source",},
						{"SourceGroup", "sourceGroup",},
						{"MktKeyword", ""},
						{"SearchKeyword", ""},
						{"Txn", ""},					
				},
				new String[][] { //pagevisit-page_visit_log]
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP", "visitip",},
						{"WebSite", "siteid",},
						{"WebPage", "",},
						{"Valid", "traffictype",},
						{"RequestUri", "requesturi",},
						{"HttpUserAgent", "httpuseragent",},
						{"HttpReferer", "httpreferer",},
						{"HttpStatus", "httpstatus",},
						{"PageVisitOrder", "pagevisitorder",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"ProductID", "productid",},
						{"BrandID",""}, 
						{"StoreID",""}, 
						{"Attribute1ID",""}, 
						{"Attribute2ID",""}, 
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"UniqueRandStr",""},
				},
				new String[][] { // outgoing-sponsor_outgoing_log
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"BusinessType", "sponsortype",},
						{"ChannelID", "channelid",},
						{"CategoryID",""}, 
						{"ProductID",""}, 
						{"MerchantID",""}, 
						{"OfferID",""}, 
						{"OfferPrice",""}, 
						{"CouponID",""}, 
						{"ZipCode",""}, 
						{"TestKey",""}, 
						{"UserID",""}, 
						{"Source",""}, 
						{"SourceGroup",""}, 
						{"MktKeyword",""}, 
						{"SearchKeyword", "keyword",},
						{"ChannelTag", "channeltag",},
						{"Rank", "displayPosition",},
						{"BidPosition",""}, 
						{"TotalMerchantCount",""},
						{"DisplayPosition", "displayPosition",},
						{"SortBy",""}, 
						{"PriceRank",""}, 
						{"RateRank",""}, 
						{"Revenue", "revenue",},
						{"RevenueLogo",""},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"OutgoingId",""}, 
						{"DestUrl", "desturl",},
						{"DisplayUrl", "advertiserHost",},
				},
				new String[][] {// outgoing-offer_outgoing_log
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},					
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"BusinessType", "businesstype",},
						{"ChannelID", "channelid",},
						{"CategoryID",""},
						{"ProductID", "productid",},
						{"MerchantID", "merchantid",},
						{"OfferID", "offerid",},
						{"OfferPrice",""}, 
						{"CouponID",""}, 
						{"ZipCode",""}, 
						{"TestKey",""}, 
						{"UserID",""}, 
						{"Source",""}, 
						{"SourceGroup",""}, 
						{"MktKeyword",""}, 
						{"SearchKeyword",""}, 
						{"ChannelTag",""}, 
						{"Rank",""}, 
						{"BidPosition", "bidposition",},
						{"TotalMerchantCount", "totalmerchantcount",},
						{"DisplayPosition", "displayposition",},
						{"SortBy", "sortby",},
						{"PriceRank", "pricerank",},
						{"RateRank", "raterank",},
						{"Revenue", "revenue",},
						{"RevenueLogo","cpcforlogo"},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"OutgoingId", "outgoingid",},
						{"DestUrl", "desturl",},
						{"DisplayUrl",""},
				},
				new String[][] { //[outgoing-affiliate_outgoing_log
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"BusinessType", "destsite",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"ProductID", "productid",},
						{"MerchantID", "merchantid",},
						{"OfferID",""}, 
						{"OfferPrice",""}, 
						{"CouponID",""}, 
						{"ZipCode",""}, 
						{"TestKey",""}, 
						{"UserID ",""}, 
						{"Source",""}, 
						{"SourceGroup",""}, 
						{"MktKeyword",""}, 
						{"SearchKeyword",""}, 
						{"ChannelTag",""}, 
						{"Rank",""}, 
						{"BidPosition",""}, 
						{"TotalMerchantCount",""}, 
						{"DisplayPosition",""}, 
						{"SortBy",""}, 
						{"PriceRank",""}, 
						{"RateRank",""}, 
						{"Revenue", "revenue",},
						{"RevenueLogo",""},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"OutgoingId",""},
						{"DestUrl", "desturl",},
						{"DisplayUrl",""},
				},
				new String[][] { //[search-search_log]"
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"Valid", "valid",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"Source", "source",},
						{"SearchKeyword", "keyword",},
						{"MatchKeyword", "matchkeyword",},
						{"IsRealSearch", "isrealsearch",},
						{"SearchEngineType", "searchenginetype",},
						{"ResultCount", "resultcount",},
						{"CostTime", "costtime",},
						{"TotalCostTime", "totalcosttime",},
						{"ResponseTime", "responsetime",},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"UniqueRandStr",""},
						{"isCached", "iscached",},
						{"ResultType", "resulttype",},
						{"ProductId", "productid",},
				},
				new String[][] {//search-affiliate_transfer_log]
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"Valid", "valid",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"Source", "source",},
						{"SearchKeyword", "keyword",},
						{"MatchKeyword",""},
						{"IsRealSearch", "isrealsearch",},
						{"SearchEngineType",""},
						{"ResultCount", "resultcount",},
						{"CostTime", "costtime",},
						{"TotalCostTime", "totalcosttime",},
						{"ResponseTime", "responsetime",},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"UniqueRandStr",""},
						{"isCached", "iscached",},
						{"ResultType", "resulttype",},
						{"ProductId", "productid",},
				},
				new String[][] { //sponsortransfer-sponsor_transfer_log]",
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid", "valid",},
						{"Keyword", "keyword",},
						{"ResultCount", "resultcount",},
						{"RequestIP", "requestip",},
						{"SponsorType", "sponsortype",},
						{"ChannelID", "channelid",},
						{"CostTime", "costtime",},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"ChannelTag", "channeltag",},
						{"WebPage",""},
						{"RequestCount", "requestcount",},
				},
				new String[][]{// impression-sponsor_impression_log]",},
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"ImpressionType", "'SL'",},
						{"BidPosition",""}, 
						{"ChannelID", "channelid",},
						{"CategoryID",""},
						{"ProductID",""},
						{"MerchantID",""},
						{"OfferID",""},
						{"ShowArea",""},
						{"SponsorType", "sponsortype",},
						{"ChannelTag", "channeltag",},
						{"SearchKeyword", "keyword",},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"ResultCount", "impressionCount",},
						{"DataSource",""},
						{"UniqueRandStr",""},
						{"PageNum",""},
				},
				new String[][] { // impression-product_impression_log]
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"ImpressionType", "impressiontype",},
						{"BidPosition", "displayposition",},
						{"ChannelID", "channelid",},
						{"CategoryID", "categoryid",},
						{"ProductID", "productid",},
						{"MerchantID", "merchantid",},
						{"OfferID", "offerid",},
						{"ShowArea",""},
						{"SponsorType",""},
						{"ChannelTag",""},
						{"SearchKeyword",""},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"ResultCount",""},
						{"DataSource",""},
						{"UniqueRandStr",""},
						{"PageNum", "pagenum",},
				},
				new String[][] {//impression-affiliate_impression_log]",},
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"ImpressionType", "'OFFER'",},
						{"BidPosition", "displayposition",},
						{"ChannelID", "channelid",},
						{"CategoryID",""},
						{"ProductID", "productid",},
						{"MerchantID", "merchantid",},
						{"OfferID", "offerid",},
						{"ShowArea", "showarea",},
						{"ResultCount",""},
						{"DataSource",""},
						{"UniqueRandStr",""},
						{"PageNum",""},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"ResultCount",""},
						{"DataSource",""},
						{"UniqueRandStr",""},
						{"PageNum",""},
				},
				new String[][] { // impression-offer_impression_log
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"VisitIP",""},
						{"WebSite", "siteid",},
						{"WebPage",""},
						{"Valid",""},
						{"ImpressionType", "'OFFER'",},
						{"BidPosition", "bidposition",},
						{"ChannelID", "channelid",},
						{"CategoryID",""},
						{"ProductID", "productid",},
						{"MerchantID", "merchantid",},
						{"OfferID", "offerid",},
						{"ShowArea", "showarea",},
						{"SponsorType",""},
						{"ChannelTag",""},
						{"SearchKeyword",""},
						{"RandStr", "randstr",},
						{"CurRandStr", "currandstr",},
						{"ResultCount",""},
						{"DataSource", "datasource",},
						{"UniqueRandStr",""},
						{"PageNum",""},
				},
				new String[][] { //clientpv-client_page_visit_log 
						{"SessionID", "sessionid",},
						{"VisitTime", "visittime",},
						{"ClientTime", "clienttime",},
						{"LoadTime", "loadtime",},
						{"CookieEnabled", "cookieenabled",},
						{"JavaEnabled", "javaenabled",},
						{"JSEnabled", "jsenabled",},
						{"ScreenSolution", "screenresolution",},
						{"Timezone", "timezone",},
						{"LangSetting", "languagesetting",},
						{"FlashVersion", "flashversion",},
						{"WebSite", "siteid",},
						{"CurrandStr", "currandstr",},
				},
		};

		for(int i = 0 ;i <logDwTable.length; i ++ ) {
			FieldLogFieldMapping mapping = new FieldLogFieldMapping(logDwTable[i][1]);
			logFileTableMap.put(logDwTable[i][0], mapping);
			Map<String,String> map = new LinkedHashMap<String,String>();
			for(int j = 0; j < fieldMapping[i].length; j ++ ) {
				map.put(fieldMapping[i][j][0], fieldMapping[i][j][1]);
			}
			mapping.putAll(map);
		}
	}


	public static class TrackingMapper extends Mapper<LongWritable,Text, Text, Text> {
		Text  outputKey = new Text();		
		Text  outputValue = new Text();
		VersionedType	vt = new VersionedType();
		StringBuilder sb = new StringBuilder();

		@Override
		protected void setup(Context context) throws IOException,
		InterruptedException {
			// TODO  read the configuration from Distributed Cache:	LogFormat.ini
			super.setup(context);
		}

		void preprocess(Text line) {
			//TODO: stripcslashes:	Returns a string with backslashes stripped off. Recognizes C-like \n, \r ..., octal and hexadecimal representation. 
			//TODO: remove \t \n \r
		}
		boolean isValid(Text line ) {
			//TODO trim in PHP
			//delete all characters:
			//			" " (ASCII 32 (0x20)), an ordinary space. 
			//			"\t" (ASCII 9 (0x09)), a tab. 
			//			"\n" (ASCII 10 (0x0A)), a new line (line feed). 
			//			"\r" (ASCII 13 (0x0D)), a carriage return. 
			//			"\0" (ASCII 0 (0x00)), the NUL-byte. 
			//			"\x0B" (ASCII 11 (0x0B)), a vertical tab. 
			return !( line.getLength() == 0 || line.equals(emptyTag) || line.equals(separator)); 
		}


		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {


			preprocess(value);

			if ( !isValid(value) )
				return;

			String line = value.toString();
			String[] records = line.split(separator.toString());

			for(String record: records) {
				if( record.isEmpty() )
					continue;

				//TODO: delimeter is fixed or not?
				//splitPreserveAllTokens only support based on single char
				String[] fields = record.split(delimeter);
				if ( fields == null || fields.length < 2)  {
					continue;
				}

				String content = getTableFieleContent(fields);
				if ( content.isEmpty()) {
					continue;
				}
				outputKey.set(fields[2]);
				outputValue.set( content );
				context.write(outputKey, outputValue);
				/*				String type = fields[0];
				if ( type.equalsIgnoreCase("redirect_log") ) {
					//TODO: store the HttpStatus
				} else if ( type.equalsIgnoreCase("user_session")) {
					//TODO: store the user Id 
				}
				 */			}
		}

		String getTableFieleContent(String[] fields) {
			String type = fields[0].substring(1, fields[0].length()-1); //trim the  { }
			String version = fields[1].substring(1, fields[1].length()-1); //trim the  { }
			vt.setType(type);
			vt.setVersion(version);

			if ( sb.length() > 0 ) {
				sb.delete(0, sb.length()-1);
			}
			LogFieldList  logFileFields = logFormatMap.get(vt);
			if( null == logFileFields) {
				logger.warn("can not find mapping for version,type" + vt);
				return "";
			}

			FieldLogFieldMapping mapping = logFileTableMap.get(type.toLowerCase());
			if ( null == mapping ) {
				logger.warn("can not find file mapping for log type: " + type.toLowerCase());
				return "";
			}

			sb.append(mapping.getTableName());
			sb.append('\t');
			for(Entry<String,String> entry : mapping.entrySet() ) {
				String logField = entry.getValue();
				if ( logField.isEmpty() ) {
					sb.append('\t');
					continue;
				}
				if ( logField.charAt(0) =='\'') { //constant
					sb.append(logField);
					sb.append('\t');
					continue;
				}
				Integer index = logFileFields.get(logField);
				if ( null != index && index+2 < fields.length ) {
					sb.append( fields[index+2]);
				}
				sb.append('\t');
			}
			return sb.toString(); 
		}
	}
	/*	public static class TrackingReducer extends Reducer<NullWritable, Text, NullWritable, Text> {

		@Override
		protected void cleanup(
				TrackingReducer.Context context)
						throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			super.cleanup(context);
		}


		@Override
		protected void setup(TrackingReducer.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			super.setup(context);
		}

		@Override
		protected void reduce(NullWritable arg0, Iterable<Text> arg1,
				TrackingReducer.Context arg2)
						throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			super.reduce(arg0, arg1, arg2);
		}

	}
	 */
	public static void main(String[] args) {
		if ( args.length != 2 ) {
			System.err.println("Usage: filePath outputPath");
			return ;
		}

		Configuration conf  = new Configuration();

		Job job;
		try {
			job = new Job(conf);
			job.setJobName("SimpleTracking");
			job.setJarByClass(SimpleTracking.class);

			job.setInputFormatClass(TextInputFormat.class);
			Path inputPath = new Path(args[0]);
			FileInputFormat.addInputPath(job, inputPath); //Add a Path to the list of inputs for the map-reduce job.

			Path outputPath = new Path(args[1]);
			FileOutputFormat.setOutputPath(job, outputPath);

			job.setOutputFormatClass(TextOutputFormat.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);

			job.setMapperClass(TrackingMapper.class);
			job.setNumReduceTasks(0);
			//			job.setReducerClass(TrackingReducer.class);

			job.waitForCompletion(true);

		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}

}
